package com.ji.storm.TestStorm.bolt;

import java.util.Map;

import org.apache.log4j.Logger;

import com.ji.storm.TestStorm.cons.Const;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class SplitBolt extends BaseRichBolt {

	private OutputCollector collector;

	public static final Logger logger = Logger.getLogger(SplitBolt.class);

	/**
	 * data prepare
	 */
	@Override
	@SuppressWarnings("rawtypes")
	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
		this.collector = collector;
	}

	@Override
	public void execute(Tuple input) {
		String words = input.getStringByField(Const.SENTENCE);
		String[] word = words.split(" ");
		for (String w : word) {
			collector.emit(new Values(w));
		}
		logger.info("####");
		// TestStringWriter.writeToTxt("##############");
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields(Const.WORD));
	}

}
